package app.dwd;

import app.BaseSqlApp;
import common.Constant;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import util.SQLUtil;

public class DwdCourseComment extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdCourseComment().init(
                3008,
                3,
                "DwdCourseComment"
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        // 1. 读取ods_db
        readOdsDb(tEnv,"DwdCourseComment");

        // 2. 过滤课程评价表
        Table reviewInfo = tEnv.sqlQuery(
                "select " +
                        "data['id'] id, " +
                        "data['user_id'] user_id, " +
                        "data['review_stars'] review_stars, " +
                        "ts ts " +
                        "from ods_db " +
                        "where `database`='edu' " +
                        "and `table`='review_info' " +
                        "and `type`='insert' ");
        tEnv.createTemporaryView("review_info", reviewInfo);

        // 3. 写出到kafka中
        tEnv.executeSql(
                "create table dwd_course_comment( " +
                        "id string, " +
                        "user_id string, " +
                        "review_stars string, " +
                        "ts bigint " +
                        ")" + SQLUtil.getKafkaSink(Constant.TOPIC_DWD_COURSE_COMMENT)
        );

        reviewInfo.executeInsert("dwd_course_comment");
    }
}
